--[[

    Filename:	auxrpc.lua

    Programmer:	aesiot

    Remark:

--]]

local auxlib = require( "auxlib" )

local auxrpc = {}

--------------------------------------------------------------------------------

local AsyncFun = nil

local Registry = {}

local Timedout = nil

local function type2char( v )

	local t = type( v )

	if "boolean" == t then

		return 'b'

	elseif "function" == t then

		return 'f'

	elseif "nil" == t then

		return 'z'

	elseif "number" == t then

		return 'n'

	elseif "string" == t then

		return 's'

	elseif "table" == t then

		return 't'

	elseif "thread" == t then

		return 'x'

	elseif "userdata" == t then

		return 'u'
	end
end

local function argfilter( param )

	local value = { t = type2char( param ), v = param }

	if 'f' == value.t then

		value.v = tostring( param )

		Registry[ value.v ] = param

	elseif 'x' == value.t or 'u' == value.t then

		value.v = null
	end

	return value
end

local function varg2list( ... )

	local args = { ... }

	local imax = auxlib.array_max_index( args ) or 0

	local result = {}

	for index = 1, imax do

		table.insert( result, argfilter( args[ index ] ) )
	end

	return result
end

--------------------------------------------------------------------------------

function auxrpc.async()

	if auxlib.is_function( AsyncFun ) then

		return AsyncFun()
	else

		return nil
	end
end

--------------------------------------------------------------------------------

function auxrpc.create_context( sendfn )

	auxlib.argcheck( auxlib.is_function( sendfn ), 1, "(function expected)" )

	local result = {}

	local metatable = { registry = {} }

	metatable.__index = function( self, index )

		if "export" == index or "import" == index or "scheduler" == index then

			return metatable[ index ]
		end

		error( "cannot get a protected members", 2 )
	end

	metatable.__metatable = "cannot get a protected metatable"

	metatable.__newindex = function( self, index, value )

		error( "cannot set a protected members", 2 )
	end

	metatable.export = function( self, name, proc )

		auxlib.argcheck( result == self, 1, "(object invalid)" )

		auxlib.argcheck( auxlib.is_string( name ), 2, "(string expected)" )

		auxlib.argcheck( auxlib.is_function( proc ) or auxlib.is_nil( proc ), 3, "(function expected)" )

		metatable.registry[ name ] = proc

		return self
	end

	metatable.import = function( self, name )

		auxlib.argcheck( result == self, 1, "(object invalid)" )

		return function( ackfn, ... )

			auxlib.argcheck( auxlib.is_function( ackfn ) or auxlib.is_nil( ackfn ), 1, "(function expected)" )

			local args = varg2list( ... )

			return auxlib.pcall( sendfn, auxlib.if_else( ackfn, function( r, c )

				auxlib.assert( auxlib.is_table( r ), "table expected" )

				auxlib.pcall( ackfn, auxlib.if_else( 'f' == r.t, self:import( r.v ), r.v ), c )

			end, nil ), name, args )
		end
	end

	metatable.scheduler = function( self, name, args, asyncfn )

		auxlib.argcheck( result == self, 1, "(object invalid)" )

		auxlib.argcheck( auxlib.is_string( name ), 2, "(string expected)" )

		auxlib.argcheck( auxlib.is_table( args ), 3, "(table expected)" )

		auxlib.argcheck( auxlib.is_function( asyncfn ) or auxlib.is_nil( asyncfn ), 4, "(function expected)" )

		local proc = metatable.registry[ name ] or Registry[ name ]

		if not proc then

			return false
		end

		local params = {}

		for i, v in ipairs( args ) do

			if 'f' == v.t then

				params[ i ] = self:import( v.v )
			else

				params[ i ] = v.v
			end
		end

		AsyncFun = auxlib.if_else( asyncfn, function()

			local fn = asyncfn()

			return function( r )

				auxlib.pcall( fn, argfilter( r ) )
			end

		end, nil )

		local b, r = auxlib.pcall( proc, table.unpack( params, 1, #args ) )

		AsyncFun = nil

		return b, argfilter( r )
	end

	setmetatable( result, metatable )

	return result
end

--------------------------------------------------------------------------------

function auxrpc.export( name, proc )

	auxlib.argcheck( auxlib.is_string( name ), 1, "(string expected)" )

	auxlib.argcheck( auxlib.is_function( proc ) or auxlib.is_nil( proc ), 2, "(function expected)" )

	Registry[ name ] = proc

	return auxrpc
end

--------------------------------------------------------------------------------

function auxrpc.export_message( vmname, method, errval, alias )

	local auxmsg = auxlib.assert( require( "auxmsg" ) )

	auxlib.argcheck( auxlib.is_string( vmname ) or auxlib.is_nil( vmname ), 1, "(string expected)" )
	auxlib.argcheck( auxlib.is_string( method ),                            2, "(string expected)" )
	auxlib.argcheck( auxlib.is_string( alias  ) or auxlib.is_nil( alias  ), 4, "(string expected)" )

	return auxrpc.export( alias or method, function( ... )

		local reply = auxrpc.async()

		if not auxlib.pcall( auxmsg.send, vmname, reply, method, ... ) then

			reply( errval )
		end
	end )
end

--------------------------------------------------------------------------------

function auxrpc.get_timeout()

	return Timedout
end

--------------------------------------------------------------------------------

function auxrpc.none( v )

	return auxlib.if_else( null == v, nil, v )
end

--------------------------------------------------------------------------------

function auxrpc.set_timeout( msec )

	auxlib.argcheck( auxlib.is_nil( msec ) or auxlib.is_integer( msec ), 1, "(integer expected)" )

	Timedout = msec
end

--------------------------------------------------------------------------------

local TaskRegistry = {}

function auxrpc.task_rpc_import( vmname )

	local auxmsg = auxlib.assert( require( "auxmsg" ) )

	auxlib.argcheck( auxlib.is_string( vmname ) or auxlib.is_nil( vmname ), 1, "(string expected)" )

	if nil == vmname then

		vmname = vm.get_task_name()
	end

	local result = TaskRegistry[ vmname ]

	if not result then

		result = auxrpc.create_context( function( ackfn, name, args )

			return auxmsg.send( vmname, function( r )

				auxlib.pcall( ackfn, r, 0 )

			end, "invoke", name, args )
		end )

		TaskRegistry[ vmname ] = result
	end

	return result
end

function auxrpc.task_rpc_init()

	local auxmsg = auxlib.assert( require( "auxmsg" ) )

	auxlib.assert( not auxmsg.exist( "invoke" ) )

	auxmsg.bind( "invoke", function( name, args )

		local rpc = auxrpc.task_rpc_import( auxmsg.get_sender() )

		local b, r = rpc:scheduler( name, args, auxmsg.async )

		return ( b and r ) or nil
	end )
end

--------------------------------------------------------------------------------

function auxrpc.mqtt_rpc_active( mqtt, idcard, crypto )

	local JSON   = auxlib.assert( require( "JSON"   ) )
	local MQTT   = auxlib.assert( require( "MQTT"   ) )
	local Crypto = auxlib.assert( require( "Crypto" ) )
	local auxtmr = auxlib.assert( require( "auxtmr" ) )

	if not auxlib.is_table( crypto ) then

		crypto = {

			decrypt = function( data ) return data end,
			encrypt = function( data ) return data end
		}
	end

	local object = { Ident = 0, RPCTable = {} }

	object.Registry = auxtmr.create_timeout_container( 1000, 5000, function( ackfn )

		ackfn()
	end )

	function object:gen_token()

		local ident = self.Ident

		self.Ident = self.Ident + 1

		return string.pack( ">j", Crypto.fnv_hash( tostring( idcard ) .. tostring( ident ) .. tostring( os.get_tick_count() ) ) )
	end

	function object:inactive()

		self.Registry.clear( true )

		self.Registry.close()

		mqtt:bind( string.format( "/rpc/%s/invoke", idcard ), nil )
		mqtt:bind( string.format( "/rpc/%s/result", idcard ), nil )

		self.Ident    = nil
		self.Registry = nil
		self.RPCTable = nil

		mqtt   = nil
		idcard = nil
		crypto = nil
		object = nil
	end

	function object:import( addr )

		auxlib.argcheck( auxlib.is_string( addr ), 2, "(string expected)" )

		local result = self.RPCTable[ addr ]

		if not result then

			result = auxrpc.create_context( function( ackfn, name, args )

				if ackfn then

					self:request( addr, name, args, function( c, r )

						if 0 == c then

							ackfn( r, 0 )

						elseif nil == c then

							ackfn( {}, -1 )
						else

							ackfn( {}, c )
						end

					end, Timedout )
				else

					self:request( addr, name, args )
				end
			end )

			self.RPCTable[ addr ] = result
		end

		return result
	end

	function object:on_topic_invoke( payload )

		local json = JSON.decode( crypto.decrypt( payload ) )

		auxlib.assert( auxlib.is_table( json ) )

		local rpc = self:import( json.c )

		local sync = true

		self.PeerName = json.c

		local b, r = rpc:scheduler( json.n, json.p, function()

			sync = false

			return function( r )

				self:response( json.c, json.k, 0, r )
			end
		end )

		self.PeerName = nil

		if sync then

			self:response( json.c, json.k, auxlib.if_else( b, 0, 500 ), r )
		end
	end

	function object:on_topic_result( payload )

		local json = JSON.decode( crypto.decrypt( payload ) )

		auxlib.assert( auxlib.is_table( json ) )

		auxlib.pcall( self.Registry.remove( json.k ), json.c, json.r )
	end

	function object:request( addr, name, args, ackfn, timeout )

		local token = self:gen_token()

		local payload = { c = idcard, n = name, p = args, k = auxlib.if_else( ackfn, token, nil ) }

		mqtt:publish( string.format( "/rpc/%s/invoke", addr ), crypto.encrypt( auxlib.to_json( payload ), addr ), MQTT.QoS0 )

		self.Registry.insert( token, ackfn, timeout )
	end

	function object:response( addr, token, code, result )

		if auxlib.is_nil( token ) then

			return
		end

		local payload = { k = token, c = code, r = result }

		mqtt:publish( string.format( "/rpc/%s/result", addr ), crypto.encrypt( auxlib.to_json( payload ), addr ), MQTT.QoS0 )
	end

	function object:getpeername()

		return self.PeerName
	end

	mqtt:observe( string.format( "/rpc/%s/invoke", idcard ), MQTT.QoS0, function( ignore, payload )

		object:on_topic_invoke( payload )
	end )

	mqtt:observe( string.format( "/rpc/%s/result", idcard ), MQTT.QoS0, function( ignore, payload )

		object:on_topic_result( payload )
	end )

	return object
end

--------------------------------------------------------------------------------

function auxrpc.mqtt_rpc_create( conf )

	auxlib.argcheck( auxlib.is_table( conf ), 1, "(table expected)" )

	auxlib.argcheck( auxlib.is_string( conf.client_id ) and 0 ~= #conf.client_id, 1, "(client_id is missing or invalid)" )

	local auxmqtt = auxlib.assert( require( "auxmqtt" ) )

	local result = {}

	local object = auxmqtt.new( conf.host, conf.port, conf.hostname, conf.cert, conf.key, conf.ca )

	object.__index = {}

	object.__metatable = "cannot get a protected metatable"

	function object:on_ready()

		self.on_ready = nil

		self:login( conf.keep_alive, conf.client_id, conf.username, conf.password, conf.will_topic, conf.will_message )
	end

	function object:on_connected()

		self.on_connected = nil

		self.RpcHook = auxrpc.mqtt_rpc_active( self, conf.client_id, conf.crypto )

		auxlib.pcall( result.on_ready, result )
	end

	function object:on_disconnect()

		self.on_disconnect = nil

		self.RpcHook:inactive()

		self.RpcHook = nil

		auxlib.pcall( result.on_close, result )
	end

	function object:on_reconnect()

		self.on_reconnect = nil

		auxlib.pcall( result.on_reconnect, result )

		object = nil
		result = nil
	end

	function object.__index:close()

		auxlib.assert( self == result )

		object:close()

		return self
	end

	function object.__index:disconnect()

		auxlib.assert( self == result )

		object:logout()

		return self
	end

	function object.__index:export( idcard, method, proc )

		auxlib.assert( self == result )

		local rpc = object.RpcHook:import( idcard )

		rpc:export( method, proc )

		return self
	end

	function object.__index:getpeername()

		auxlib.assert( self == result )

		return object.RpcHook:getpeername()
	end

	function object.__index:import( idcard, method )

		auxlib.assert( self == result )

		local rpc = object.RpcHook:import( idcard )

		return rpc:import( method )
	end

	setmetatable( result, object )

	return result
end

--------------------------------------------------------------------------------

function auxrpc.udp_rpc_create( addr, crypto )

	local JSON   = auxlib.assert( require( "JSON"   ) )
	local Crypto = auxlib.assert( require( "Crypto" ) )
	local auxsck = auxlib.assert( require( "auxsck" ) )
	local auxtmr = auxlib.assert( require( "auxtmr" ) )

	if not auxlib.is_table( crypto ) then

		crypto = {

			decrypt = function( data ) return data end,
			encrypt = function( data ) return data end
		}
	end

	local result = {}

	local object = auxsck.new_udp_object( addr )

	object.__index = {}

	object.__metatable = "cannot get a protected metatable"

	object.ReqIdent = 0

	object.RPCTable = {}

	object.Registry = auxtmr.create_timeout_container( 1000, 5000, function( ackfn )

		ackfn()
	end )

	function object:on_close()

		self.on_close = nil

		self.Registry.clear( true )

		self.Registry.close()

		auxlib.pcall( result.on_close, result )

		self.ReqIdent = nil
		self.RPCTable = nil
		self.Registry = nil

		object = nil
		result = nil
		crypto = nil
	end

	function object:on_ready()

		self.on_ready = nil

		auxlib.pcall( result.on_ready, result )
	end

	function object:on_read( data, addr )

		local json = JSON.decode( crypto.decrypt( data, addr ) )

		auxlib.assert( auxlib.is_table( json ) )

		if "req" == json.m then

			local rpc = self:import( addr )

			local sync = true

			self.PeerName = addr

			local b, r = rpc:scheduler( json.n, json.p, function()

				sync = false

				return function( r )

					self:response( addr, json.k, 0, r )
				end
			end )

			self.PeerName = nil

			if sync then

				self:response( addr, json.k, auxlib.if_else( b, 0, 500 ), r )
			end

		elseif "ret" == json.m then

			auxlib.pcall( self.Registry.remove( json.k ), json.c, json.r )
		end
	end

	function object:gen_token()

		local ident = self.ReqIdent

		self.ReqIdent = self.ReqIdent + 1

		return string.pack( ">j", Crypto.fnv_hash( tostring( addr ) .. tostring( ident ) .. tostring( os.get_tick_count() ) ) )
	end

	function object:import( addr )

		auxlib.argcheck( auxlib.is_string( addr ), 2, "(string expected)" )

		local result = self.RPCTable[ addr ]

		if not result then

			result = auxrpc.create_context( function( ackfn, name, args )

				if ackfn then

					self:request( addr, name, args, function( c, r )

						if 0 == c then

							ackfn( r, 0 )

						elseif nil == c then

							ackfn( {}, -1 )
						else

							ackfn( {}, c )
						end

					end, Timedout )
				else

					self:request( addr, name, args )
				end
			end )

			self.RPCTable[ addr ] = result
		end

		return result
	end

	function object:request( addr, name, args, ackfn, timeout )

		local token = self:gen_token()

		local data = { m = "req", n = name, p = args, k = auxlib.if_else( ackfn, token, nil ) }

		self:send( addr, crypto.encrypt( auxlib.to_json( data ), addr ) )

		self.Registry.insert( token, ackfn, timeout )
	end

	function object:response( addr, token, code, result )

		if auxlib.is_nil( token ) then

			return
		end

		local data = { m = "ret", k = token, c = code, r = result }

		self:send( addr, crypto.encrypt( auxlib.to_json( data ), addr ) )
	end

	function object.__index:close()

		auxlib.assert( self == result )

		object:close()

		return self
	end

	function object.__index:export( addr, name, proc )

		auxlib.assert( self == result )

		local rpc = object:import( addr )

		rpc:export( name, proc )

		return self
	end

	function object.__index:getpeername()

		auxlib.assert( self == result )

		return object.PeerName
	end

	function object.__index:import( addr, name )

		auxlib.assert( self == result )

		local rpc = object:import( addr )

		return rpc:import( name )
	end

	setmetatable( result, object )

	return result
end

--------------------------------------------------------------------------------

return auxrpc
